package pkg

import (
	"common/global"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/google/uuid"
	"github.com/gorilla/websocket"
	"log"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

// 全局变量
var ( // Redis 客户端
	upgrader    = websocket.Upgrader{}             // WebSocket 升级器
	clientMutex sync.Mutex                         // 客户端映射锁
	clients     = make(map[string]*websocket.Conn) // 在线用户映射 (userID -> conn)
)

// 消息结构体
type Message struct {
	ID         string `json:"id"`          // 消息唯一ID
	SenderID   string `json:"sender_id"`   // 新增发送方ID
	ReceiverID string `json:"receiver_id"` // 新增接收方ID
	Content    string `json:"content"`     // 消息内容
	Timestamp  int64  `json:"timestamp"`   // 时间戳
	Type       string `json:"type"`        // 消息类型
}

func Chat() {
	// 初始化 Redis

	// 注册路由
	http.Handle("/send", http.HandlerFunc(handleSendMessage))
	http.HandleFunc("/ws", handleWebSocket)

	// 启动服务器
	//log.Println("Server started at :8080")
	//log.Fatal(http.ListenAndServe(":8080", nil))
	// 创建 HTTP 服务器
	srv := &http.Server{
		Addr: ":8081",
	}

	// 使用 goroutine 启动 HTTP 服务器
	go func() {
		log.Println("HTTP server started at :8081")
		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			log.Fatalf("listen: %s\n", err)
		}
	}()

	// 等待中断信号以优雅关闭服务器
	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	<-quit
	log.Println("Shutting down server...")

	// 创建5秒超时的context
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := srv.Shutdown(ctx); err != nil {
		log.Fatal("Server forced to shutdown:", err)
	}

	log.Println("Server exiting")
}

// WebSocket 处理函数
func handleWebSocket(w http.ResponseWriter, r *http.Request) {
	// 从查询参数获取用户ID
	userID := r.URL.Query().Get("user_id")
	if userID == "" {
		http.Error(w, "user_id parameter required", http.StatusBadRequest)
		return
	}

	// 升级为 WebSocket 连接
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("WebSocket upgrade failed: %v", err)
		return
	}
	defer conn.Close()

	// 注册客户端
	registerClient(userID, conn)
	defer unregisterClient(userID)

	// 发送未读消息
	sendPendingMessages(userID, conn)

	// 保持连接 (监听关闭)
	for {
		if _, _, err := conn.ReadMessage(); err != nil {
			break
		}
	}
}

// 注册客户端
func registerClient(userID string, conn *websocket.Conn) {
	clientMutex.Lock()
	defer clientMutex.Unlock()
	clients[userID] = conn
	log.Printf("User %s connected", userID)
}

// 注销客户端
func unregisterClient(userID string) {
	clientMutex.Lock()
	defer clientMutex.Unlock()
	delete(clients, userID)
	log.Printf("User %s disconnected", userID)
}

// 发送待处理消息
func sendPendingMessages(userID string, conn *websocket.Conn) {
	ctx := context.Background()
	// 从离线队列获取私信
	offlineKey := fmt.Sprintf("offline:%s", userID)
	offlineMsgs, _ := global.Client.LRange(ctx, offlineKey, 0, -1).Result()
	global.Client.Del(ctx, offlineKey) // 清空离线消息

	// 发送并处理离线私信
	for _, msgJson := range offlineMsgs {
		var msg Message
		if err := json.Unmarshal([]byte(msgJson), &msg); err == nil {
			sendToConnection(conn, &msg)
		}
	}
	// 新增广播消息检查
	broadcastMsgs, _ := global.Client.LRange(ctx, "broadcast:history", 0, -1).Result()
	for _, msgJson := range broadcastMsgs {
		var msg Message
		if err := json.Unmarshal([]byte(msgJson), &msg); err == nil {
			sendToConnection(conn, &msg)
		} else {
			log.Printf("解析广播消息失败: %v", err)
		}
	}
}

// 处理消息发送请求
func handleSendMessage(w http.ResponseWriter, r *http.Request) {
	if r.Method != "POST" {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	// 解析表单参数
	msgType := r.FormValue("type")
	receiverID := r.FormValue("receiver_id")
	content := r.FormValue("content")

	// 从请求上下文获取发送者ID（需要实现认证中间件）
	senderID := r.FormValue("sender_id") // 默认系统发送
	if msgType == "private" {
		if senderID == "" {
			http.Error(w, "sender_id is required for private messages", http.StatusBadRequest)
			return
		}
		if receiverID == "" {
			http.Error(w, "receiver_id is required for private messages", http.StatusBadRequest)
			return
		}
	}
	// 创建消息
	msg := Message{
		ID:         uuid.New().String(),
		SenderID:   senderID,
		ReceiverID: receiverID,
		Content:    content,
		Timestamp:  time.Now().Unix(),
		Type:       msgType,
	}

	// 根据消息类型处理存储
	switch msgType {
	case "private":
		if err := storePrivateMessage(&msg); err != nil {
			http.Error(w, "Failed to store private message", http.StatusInternalServerError)
			return
		}
	case "broadcast":
		if err := storeBroadcastMessage(&msg); err != nil {
			http.Error(w, "存储广播消息失败", http.StatusInternalServerError)
			return
		}
	default:
		if err := storeMessage(&msg); err != nil {
			http.Error(w, "Failed to store message", http.StatusInternalServerError)
			return
		}
	}

	// 实时消息推送
	sendRealTimeMessage(&msg)

	w.WriteHeader(http.StatusCreated)
	json.NewEncoder(w).Encode(map[string]string{"message_id": msg.ID})
}

// 新增广播消息存储逻辑
func storeBroadcastMessage(msg *Message) error {
	ctx := context.Background()

	// 存储到广播历史记录
	pipe := global.Client.Pipeline()
	msgJson, _ := json.Marshal(msg)

	// 记录全局广播消息
	pipe.LPush(ctx, "broadcast:history", msgJson)

	// 为所有用户存储离线消息
	pipe.SAdd(ctx, "broadcast:users", "*") // 实际生产环境需替换为真实用户列表
	users, _ := pipe.SMembers(ctx, "broadcast:users").Result()

	for _, userID := range users {
		pipe.LPush(ctx, fmt.Sprintf("messages:%s", userID), msgJson)
		pipe.Expire(ctx, fmt.Sprintf("messages:%s", userID), 24*time.Hour)
	}

	_, err := pipe.Exec(ctx)
	return err
}

// 新增私信存储逻辑
func storePrivateMessage(msg *Message) error {
	ctx := context.Background()

	// 仅当接收方离线时存储
	clientMutex.Lock()
	_, online := clients[msg.ReceiverID]
	clientMutex.Unlock()

	if !online {
		// 存储到接收方离线队列
		key := fmt.Sprintf("offline:%s", msg.ReceiverID)
		msgJson, _ := json.Marshal(msg)
		return global.Client.RPush(ctx, key, msgJson).Err()
	}
	return nil
}

// 存储消息到 Redis
func storeMessage(msg *Message) error {
	ctx := context.Background()

	// 序列化消息
	msgJson, err := json.Marshal(msg)
	if err != nil {
		return err
	}

	// 存储到两个地方:
	// 1. 消息内容存储 (HSET)
	// 2. 用户消息列表 (LPUSH)
	userKey := fmt.Sprintf("messages:%s", msg.ReceiverID)

	pipe := global.Client.TxPipeline()
	pipe.Set(ctx, msg.ID, msgJson, 0)
	pipe.LPush(ctx, userKey, msg.ID)
	_, err = pipe.Exec(ctx)

	return err
}

// 扩展实时消息推送
func sendRealTimeMessage(msg *Message) {
	clientMutex.Lock()
	defer clientMutex.Unlock()

	switch msg.Type {
	case "private":
		if conn, ok := clients[msg.ReceiverID]; ok {
			// 接收方在线直接发送
			sendToConnection(conn, msg)
		} else {
			// 存储到离线队列
			key := fmt.Sprintf("offline:%s", msg.ReceiverID)
			msgJson, _ := json.Marshal(msg)
			global.Client.RPush(context.Background(), key, msgJson)
		}
	case "broadcast":
		// 群发所有在线用户
		for _, conn := range clients {
			sendToConnection(conn, msg)
		}
	default:
		// 系统消息默认处理
		if conn, ok := clients[msg.ReceiverID]; ok {
			sendToConnection(conn, msg)
		}
	}
}

func sendToConnection(conn *websocket.Conn, msg *Message) {
	msgJson, _ := json.Marshal(msg)
	if err := conn.WriteMessage(websocket.TextMessage, msgJson); err != nil {
		log.Printf("消息发送失败: %v", err)
	}

	// 群发消息不删除存储
	if msg.Type != "broadcast" {
		if err := deleteMessage(msg.ReceiverID, msg.ID); err != nil {
			log.Printf("删除消息失败: %v", err)
		}
	}
}

// 删除消息
func deleteMessage(userID, msgID string) error {
	ctx := context.Background()
	userKey := fmt.Sprintf("messages:%s", userID)

	pipe := global.Client.TxPipeline()
	pipe.Del(ctx, msgID)
	pipe.LRem(ctx, userKey, 1, msgID)
	_, err := pipe.Exec(ctx)

	return err
}
